"""
Case Type   : 协议兼容
Case Name   : sha2协议连接成功，index相关操作
Create At   : 2024.09.23
Owner       : @zhanghuan96
Description :
    1、创建兼容b库并查询插件是否加载成功
    2、创建用户
    3、caching_sha2认证
    4、查询系统未使用端口
    5、设置协议兼容guc参数
    6、连接og侧B库创建btree索引
    7、og数据库查询
    8、连接og侧B库创建unique索引
    9、og数据库查询
    10、连接og侧B库创建全文索引
    11、og数据库查询
    12、连接og侧B库创建hash索引
    13、og数据库查询
    14、连接og侧B库创建psort索引
    15、og数据库查询
    16、连接og侧B库创建分区表索引
    17、og数据库查询
    18、清理环境
Expect      :
    1、成功，插件查询成功
    2、成功
    3、成功
    4、成功
    5、成功
    6、成功
    7、成功
    8、成功
    9、成功
    10、成功
    11、成功
    12、成功
    13、成功
    14、成功
    15、成功
    16、成功
    17、成功
    18、成功
History     :
"""
import os
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonMySQL import CommonMySQL
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class CompatibilityTest01(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.common = Common()
        self.sh_pri = CommonSH('PrimaryDbUser')
        self.pri_node = Node('PrimaryDbUser')
        self.my_node = Node('mysql8')
        self.constant = Constant()
        self.db_name = 'db_protocol_0020'
        self.u_name = 'u_protocol_0020'
        self.tb_name = 'tb_protocol_0020'
        self.inx_name = 'inx_protocol_0020'
        self.conf_path = os.path.join(macro.DB_INSTANCE_PATH,
                                      'postgresql.conf')
        self.default_value1 = self.common.show_param('enable_dolphin_proto')
        self.default_value2 = self.common.show_param('dolphin_server_port')
        self.exec_sql1 = f'''drop table if exists {self.tb_name};
        create table {self.tb_name}(id int primary key,email varchar(50),\
        index email_index using btree(id,email));'''
        self.exec_sql2 = f'''drop table if exists {self.tb_name}_01;
        create table {self.tb_name}_01(id int primary key,email varchar(50),\
        unique key id_index(id));'''
        self.exec_sql3 = f'''drop table if exists {self.tb_name}_02;
        CREATE TABLE  {self.tb_name}_02 (id int unsigned auto_increment not \
        null primary key,title varchar, boby text, name name);
        drop index if exists {self.inx_name}_index_1;
        CREATE FULLTEXT INDEX {self.inx_name}_index_1 ON  {self.tb_name}_02 \
        (title, boby)WITH PARSER ngram;'''
        self.exec_sql4 = f'''drop table if exists {self.tb_name}_03;
        create table {self.tb_name}_03(id int,name varchar(20));
        drop index if exists idx;
        create index idx on {self.tb_name}_03 using hash(name);'''
        self.exec_sql5 = f'''drop table if exists {self.tb_name}_04;
        create table {self.tb_name}_04(id int,name varchar(20))\
        with(orientation=column);
        drop index if exists idx;
        create index idx on {self.tb_name}_04 using psort(name);'''
        self.exec_sql6 = f'''drop table if exists {self.tb_name}_05;
        create table {self.tb_name}_05(id int, name varchar(20),score int)\
        partition by range(score)(partition p1 values less than(60),\
        partition p2 values less than(100));
        drop index if exists {self.inx_name};
        create index {self.inx_name} on {self.tb_name}_05(score) local;
        drop index if exists {self.inx_name}_01;
        create index {self.inx_name}_01 on {self.tb_name}_05(id) global;'''

    def test_create_function(self):
        text = '-----step1:创建兼容b库并查询插件;expect:成功-----'
        self.log.info(text)
        sql_cmd = (f"drop database if exists {self.db_name};"
                   f"create database {self.db_name} encoding 'utf8' "
                   f"dbcompatibility ='B';")
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd)
        self.log.info(sql_res)
        self.assertEqual(1,
                         sql_res.count(self.constant.CREATE_DATABASE_SUCCESS),
                         '兼容B库创建失败')
        sql_cmd1 = "select extname from pg_extension where extname ='dolphin';"
        self.log.info(sql_cmd1)
        sql_res1 = self.sh_pri.execut_db_sql(sql_cmd1, dbname=self.db_name)
        self.log.info(sql_res1)
        self.assertIn('dolphin', sql_res1, '查询失败')

        text = '---step2:创建用户;expect:成功---'
        self.log.info(text)
        sql_cmd2 = (f"drop user if exists {self.u_name};"
                    f"create user {self.u_name} with password "
                    f"'{macro.COMMON_PASSWD}';"
                    f"grant all privileges to {self.u_name};")
        self.log.info(sql_cmd2)
        sql_res2 = self.sh_pri.execut_db_sql(sql_cmd2, dbname=self.db_name)
        self.log.info(sql_res2)
        self.assertIn('CREATE ROLE', sql_res2, '创建失败')
        self.assertIn('ALTER ROLE', sql_res2, '创建失败')

        text = '---step3:caching_sha2认证;expect:成功---'
        self.log.info(text)
        sql_cmd = (f"select set_caching_sha2_password('{self.u_name}', "
                   f"'{macro.COMMON_PASSWD}', '');")
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('(1 row)', sql_res, '创建失败')

        text = '----step4:查询系统未使用端口;expect:成功----'
        self.log.info(text)
        port = self.common.get_not_used_port(self.pri_node)
        self.assertNotEqual(0, port, '执行失败:' + text)

        text = '-step5:设置协议兼容guc参数;expect:成功-'
        self.log.info(text)
        msg = self.sh_pri.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        'enable_dolphin_proto = on')
        self.log.info(msg)
        self.assertTrue(msg)
        msg = self.sh_pri.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        f'dolphin_server_port ={port}')
        self.log.info(msg)
        self.assertTrue(msg)
        sed_cmd = (f"sed -i '$a dolphin.default_database_name={self.db_name}' "
                   f"{self.conf_path}")
        self.log.info(sed_cmd)
        res = self.common.get_sh_result(self.sh_pri.node, sed_cmd)
        self.log.info(res)
        msg = self.sh_pri.restart_db_cluster()
        self.log.info(msg)
        status = self.sh_pri.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status)

        text = '-step6:连接og侧B库创建btree索引;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql1, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertNotIn('ERROR', c_res)

        text = '-step7:og数据库查询;expect:成功--'
        self.log.info(text)
        sql_cmd = f"show index from {self.tb_name} in  {self.u_name};"
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('| email_index           |            1 | id',
                      sql_res)
        self.assertIn('| email_index           |            2 | email',
                      sql_res)
        self.assertIn('| tb_protocol_0020_pkey |            1 | id',
                      sql_res)

        text = '-step8:连接og侧B库创建unique索引;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql2, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertNotIn('ERROR', c_res)

        text = '-step9:og数据库查询;expect:成功--'
        self.log.info(text)
        sql_cmd = f"show index from {self.tb_name}_01 in  {self.u_name};"
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('| id_index |            1 | id ', sql_res)

        text = '-step10:连接og侧B库创建全文索引;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql3, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertNotIn('ERROR', c_res)

        text = '-step11:og数据库查询;expect:成功--'
        self.log.info(text)
        sql_cmd = f"\d+ {self.u_name}.{self.tb_name}_02;"
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('"inx_protocol_0020_index_1" gin '
                      '(to_tsvector(\'ngram\'::regconfig, title::text), '
                      'to_tsvector(\'ngram\'::regconfig, boby))', sql_res)

        text = '-step12:连接og侧B库创建hash索引;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql4, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertNotIn('ERROR', c_res)

        text = '-step13:og数据库查询;expect:成功--'
        self.log.info(text)
        sql_cmd = f"show index from {self.tb_name}_03 in  {self.u_name};"
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('tb_protocol_0020_03 | t          | idx     '
                      ' |            1 | name',
                      sql_res)

        text = '-step14:连接og侧B库创建psort索引;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql5, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertNotIn('ERROR', c_res)

        text = '-step15:og数据库查询;expect:成功--'
        self.log.info(text)
        sql_cmd = f"show index from {self.tb_name}_04 in  {self.u_name};"
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('tb_protocol_0020_04 | t          | idx      '
                      '|            1 | name ', sql_res)

        text = '-step16:连接og侧B库创建分区表索引;expect:成功--'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.exec_sql6, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertNotIn('ERROR', c_res)

        text = '-step17:og数据库查询;expect:成功--'
        self.log.info(text)
        sql_cmd = f"show index from {self.tb_name}_05 in  {self.u_name};"
        self.log.info(sql_cmd)
        sql_res = self.sh_pri.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.log.info(sql_res)
        self.assertIn('| inx_protocol_0020    |            1 | score',
                      sql_res)
        self.assertIn('| inx_protocol_0020_01 |            1 | id ',
                      sql_res)

    def tearDown(self):
        text = '-----step18:清理环境;expect:成功-----'
        self.log.info(text)
        drop_cmd = f"drop database if exists {self.db_name};"
        self.log.info(drop_cmd)
        drop_res = self.sh_pri.execut_db_sql(drop_cmd, dbname='postgres')
        self.log.info(drop_res)
        drop_cmd1 = f"drop user if exists {self.u_name};"
        self.log.info(drop_cmd1)
        drop_res1 = self.sh_pri.execut_db_sql(drop_cmd1)
        self.log.info(drop_res1)

        guc_cmd = self.sh_pri.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'enable_dolphin_proto = '
                                            f'{self.default_value1}')
        self.log.info(guc_cmd)
        self.assertTrue(guc_cmd)
        guc_cmd = self.sh_pri.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'dolphin_server_port ='
                                            f'{self.default_value2}')
        self.log.info(guc_cmd)
        self.assertTrue(guc_cmd)

        sed_cmd = (f"sed -i '/dolphin.default_database_name={self.db_name}/d' "
                   f"{self.conf_path}")
        self.log.info(sed_cmd)
        res = self.common.get_sh_result(self.sh_pri.node, sed_cmd)
        self.log.info(res)
        msg = self.sh_pri.restart_db_cluster()
        self.log.info(msg)
        status = self.sh_pri.get_db_cluster_status()
        self.log.info(status)
        self.assertTrue("Degraded" in status or "Normal" in status)
        self.assertIn('DROP ROLE', drop_res1, '清理环境失败')
        self.assertEqual(1,
                         drop_res.count(self.constant.DROP_DATABASE_SUCCESS),
                         '清理环境失败')
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
